package com.shujia.tf

import java.sql.Connection

import org.apache.flink.api.common.functions.RichMapFunction
import com.shujia.bean.{Car, CarWide}
import com.shujia.util.JDBCUtil
import org.apache.flink.configuration.Configuration

class CarJoinKcMap extends RichMapFunction[Car, CarWide] {

  var con: Connection = _

  override def open(parameters: Configuration): Unit = {
    con = JDBCUtil.getConnection
  }


  override def close(): Unit = {
    con.close()
  }

  override def map(car: Car): CarWide = {

    val carWide = new CarWide()
    carWide.setCarId(car.getCarId)
    carWide.setSpeed(car.getSpeed)
    carWide.setTs(car.getTs)

    //通过卡口编号查询卡口信息
    val kcId = car.getKcId
    val sql = "select * from kc where id=?"
    val stat = con.prepareStatement(sql)
    stat.setString(1, kcId)
    val result = stat.executeQuery()

    if (result.next()) {
      val dlId = result.getString("dl_id")
      val quId = result.getString("qu_id")
      val rateLimiting = result.getString("rate_limiting")

      carWide.setRateLimiting(rateLimiting.toLong)

      //关联道路表区域表
      val dlstat = con.prepareStatement("select * from dl where id=?")
      dlstat.setString(1, dlId)
      val dlresult = dlstat.executeQuery()
      if (dlresult.next()) {
        val dlName = dlresult.getString("name")
        carWide.setDlName(dlName)
      }

      val quStat = con.prepareStatement("select * from qu where id=?")
      quStat.setString(1, quId)

      val quResult = quStat.executeQuery()
      if (quResult.next()) {
        val city = quResult.getString("city")
        val county = quResult.getString("county")

        carWide.setCity(city)
        carWide.setCounty(county)
      }
    }


    //返回结果
    carWide
  }
}
